package com.wafersystems.cloud.configuration;

import org.eclipse.paho.client.mqttv3.MqttClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.dsl.IntegrationFlow;
import org.springframework.integration.dsl.IntegrationFlows;
import org.springframework.integration.endpoint.MessageProducerSupport;
import org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory;
import org.springframework.integration.mqtt.core.MqttPahoClientFactory;
import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.MessagingException;
import org.springframework.stereotype.Component;

import com.alibaba.fastjson.JSON;
import com.wafersystems.cloud.bo.MyProps;
import com.wafersystems.cloud.service.InterfaceMsgNotice;
import com.wafersystems.cloud.vo.MqMsg;

import lombok.extern.slf4j.Slf4j;

@Component
@Configuration
@Slf4j
public class NjwlMqttListener {
  @Autowired
  private MyProps myProps;
  @Autowired
  private InterfaceMsgNotice notice;

  @Bean
  public MqttPahoClientFactory mqttClientFactory2() {
    DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
    factory.setServerURIs(
        "tcp://" + myProps.getMyMqtt().getHost() + ":" + myProps.getMyMqtt().getPort());
    factory.setUserName(myProps.getMyMqtt().getUsername());
    factory.setPassword(myProps.getMyMqtt().getPassword());
    return factory;
  }

  @Bean
  public IntegrationFlow mqttInFlow() {
    return IntegrationFlows.from(mqttInbound()).transform(p -> p).handle(handler()).get();
  }

  @Bean
  public MessageProducerSupport mqttInbound() {
    MqttPahoMessageDrivenChannelAdapter adapter = new MqttPahoMessageDrivenChannelAdapter(
        MqttClient.generateClientId(), mqttClientFactory2(), myProps.getMyMqtt().getTopic());
    adapter.setCompletionTimeout(5000);
    adapter.setQos(2);
    return adapter;
  }

  @Bean
  public MessageHandler handler() {
    return new MessageHandler() {

      @Override
      public void handleMessage(Message msgObj) throws MessagingException {
        log.info("recived msg is : " + msgObj);
        try {
          String msg = msgObj.getPayload().toString();
          MqMsg message = JSON.parseObject(msg.toString(), MqMsg.class);
          message.setSocketId(message.getSocketId()+message.getSocketAttr());
          notice.sendMsg(JSON.toJSONString(message));
        } catch (Exception e) {
          log.error(e.getMessage(), e);
        }
      }
    };
  }

}
